package com.google.code.stalkernet.async.io;

import java.io.IOException;
import java.net.Socket;
import java.net.SocketAddress;
import java.net.SocketOption;
import java.nio.ByteBuffer;
import java.nio.channels.NetworkChannel;
import java.nio.channels.SocketChannel;
import java.util.Set;

import com.google.code.stalkernet.reactor.Reactor;

/**
 * Asynchronous wrapper for {@link SocketChannel}.
 *
 * @author Mikhail Vladimirov
 */
public class AsyncSocketChannel
    extends AbstractAsyncSelectableChannel
    implements AsyncByteChannel, AsyncGatheringByteChannel,
        AsyncScatteringByteChannel, NetworkChannel
{
    private final SocketChannel socketChannel;

    private Runnable pendingConnect = null;
    private Runnable pendingRead = null;
    private Runnable pendingWrite = null;

    /**
     * Create new asynchronous socket channel wrapping given
     * {@link SocketChannel}.
     *
     * @param socketChannel {@link SocketChannel} to wrap.
     * @param reactor {@link Reactor} to use
     * @throws IOException if I/O error occurred
     */
    public AsyncSocketChannel (SocketChannel socketChannel, Reactor reactor)
        throws IOException
    {
        super (socketChannel, reactor);

        if (socketChannel == null)
            throw new IllegalArgumentException ("Socket channel is null");

        this.socketChannel = socketChannel;
    }

    /**
     * Open new asynchronous socket channel.
     *
     * @param reactor Reactor to use
     * @return {@link AsyncSocketChannel} object
     * @throws IOException if I/O error occurred
     */
    public static AsyncSocketChannel open (Reactor reactor)
        throws IOException
    {
        return new AsyncSocketChannel (SocketChannel.open (), reactor);
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncByteChannel
    public Handle read (ByteBuffer dst, Callback <Integer> callback)
    {
        if (dst == null)
            throw new IllegalArgumentException ("Dst is null");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        if (pendingRead != null)
            throw new IllegalStateException ("Pending read is not null");

        AsyncReadSingle asyncReadSingle =
            new AsyncReadSingle (dst, callback);

        pendingRead = asyncReadSingle;

        updateInterestOps ();

        return asyncReadSingle;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncByteChannel
    public Handle write (ByteBuffer src, Callback <Integer> callback)
    {
        if (src == null)
            throw new IllegalArgumentException ("Src is null");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        if (pendingWrite != null)
            throw new IllegalStateException ("Pending write is not null");

        AsyncWriteSingle asyncWriteSingle =
            new AsyncWriteSingle (src, callback);

        pendingWrite = asyncWriteSingle;

        updateInterestOps ();

        return asyncWriteSingle;
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public boolean isOpen ()
    {
        return socketChannel.isOpen ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public void close () throws IOException
    {
        socketChannel.close ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public NetworkChannel bind (SocketAddress local) throws IOException
    {
        socketChannel.bind (local);

        return this;
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public SocketAddress getLocalAddress () throws IOException
    {
        return socketChannel.getLocalAddress ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public <T> NetworkChannel setOption (SocketOption <T> name, T value)
        throws IOException
    {
        socketChannel.setOption (name, value);

        return this;
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public <T> T getOption (SocketOption <T> name) throws IOException
    {
        return socketChannel.getOption (name);
    }

    /**
     * {@inheritDoc}
     */
    @Override // NetworkChannel
    public Set <SocketOption <?>> supportedOptions ()
    {
        return socketChannel.supportedOptions ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncScatteringByteChannel
    public Handle read (ByteBuffer [] dsts, int offset, int length,
        Callback <Long> callback)
    {
        if (dsts == null)
            throw new IllegalArgumentException ("Dsts is null");

        if (offset < 0)
            throw new IllegalArgumentException ("Offset (" + offset + ") < 0");

        if (length < 0)
            throw new IllegalArgumentException ("Length (" + length + ") < 0");

        int dstsLength = dsts.length;

        if (offset + length > dstsLength)
            throw new IllegalArgumentException ("Offset (" + offset +
                ") + length (" + length + ") > dsts length (" + dstsLength + ")");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        if (pendingRead != null)
            throw new IllegalStateException ("Pending read is not null");

        AsyncReadMultiple asyncReadMultiple =
            new AsyncReadMultiple (dsts, offset, length, callback);

        pendingRead = asyncReadMultiple;

        updateInterestOps ();

        return asyncReadMultiple;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncScatteringByteChannel
    public Handle read (ByteBuffer [] dsts, Callback <Long> callback)
    {
        if (dsts == null)
            throw new IllegalArgumentException ("Dsts is null");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        return read (dsts, 0, dsts.length, callback);
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncGatheringByteChannel
    public Handle write (ByteBuffer [] srcs, int offset, int length,
        Callback <Long> callback)
    {
        if (srcs == null)
            throw new IllegalArgumentException ("Srcs is null");

        if (offset < 0)
            throw new IllegalArgumentException ("Offset (" + offset + ") < 0");

        if (length < 0)
            throw new IllegalArgumentException ("Length (" + length + ") < 0");

        int srcsLength = srcs.length;

        if (offset + length > srcsLength)
            throw new IllegalArgumentException ("Offset (" + offset +
                ") + length (" + length + ") > srcs length (" + srcsLength + ")");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        if (pendingWrite != null)
            throw new IllegalStateException ("Pending write is not null");

        AsyncWriteMultiple asyncWriteMultiple =
            new AsyncWriteMultiple (srcs, offset, length, callback);

        pendingWrite = asyncWriteMultiple;

        updateInterestOps ();

        return asyncWriteMultiple;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AsyncGatheringByteChannel
    public Handle write (ByteBuffer [] srcs, Callback <Long> callback)
    {
        if (srcs == null)
            throw new IllegalArgumentException ("Srcs is null");

        if (callback == null)
            throw new IllegalArgumentException ("Callback is null");

        return write (srcs, 0, srcs.length, callback);
    }

    /**
     * Wrapper for {@link SocketChannel#shutdownInput()}.
     *
     * @return this {@link AsyncSocketChannel} object
     * @throws IOException if I/O error occurred
     */
    public AsyncSocketChannel shutdownInput() throws IOException
    {
        socketChannel.shutdownInput ();

        return this;
    }

    /**
     * Wrapper for {@link SocketChannel#shutdownOutput()}.
     *
     * @return this {@link AsyncSocketChannel} object
     * @throws IOException if I/O error occurred
     */
    public AsyncSocketChannel shutdownOutput() throws IOException
    {
        socketChannel.shutdownOutput ();

        return this;
    }

    /**
     * Wrapper for {@link SocketChannel#socket()}.
     *
     * @return A socket associated with this channel
     */
    public Socket socket()
    {
        return socketChannel.socket ();
    }

    /**
     * Wrapper for {@link SocketChannel#isConnected()}.
     *
     * @return <tt>true</tt> if, and only if, this channel's network socket
     *         is {@link #isOpen open} and connected
     */
    public boolean isConnected()
    {
        return socketChannel.isConnected ();
    }

    /**
     * Wrapper for {@link SocketChannel#isConnectionPending()}.
     *
     * @return <tt>true</tt> if, and only if, a connection operation has been
     *         initiated on this channel but not yet completed by invoking the
     *         {@link SocketChannel#finishConnect} method
     */
    public boolean isConnectionPending()
    {
        return socketChannel.isConnectionPending ();
    }

    /**
     * Asynchronous wrapper for {@link SocketChannel#connect(SocketAddress)}.
     *
     * @param remote The remote address to which this channel is to be connected
     * @param callback callback to receive asynchronous operation result
     * @return asynchronous operation handle
     */
    public Handle connect(SocketAddress remote, final Callback <Void> callback)
    {
        if (remote == null)
            throw new IllegalArgumentException ("Remote is null");

        if (pendingConnect != null)
            throw new IllegalStateException ("Pending connect is not null");

        try
        {
            if (socketChannel.connect (remote))
            {
                callback.onSuccess (null);

                return null;
            }
            else
            {
                AsyncConnect asyncConnect = new AsyncConnect (callback);

                pendingConnect = asyncConnect;

                updateInterestOps ();

                return asyncConnect;
            }
        }
        catch (IOException ex)
        {
            callback.onError (ex);
            return null;
        }
    }

    /**
     * Wrapper for {@link SocketChannel#getRemoteAddress()}.
     *
     * @return The remote address; {@code null} if the channel's socket is not
     *         connected
     * @throws IOException if I/O error occurred
     */
    public SocketAddress getRemoteAddress() throws IOException
    {
        return socketChannel.getRemoteAddress ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected boolean wantAccept ()
    {
        return false;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected void canAccept ()
    {
        // Do nothing
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected boolean wantConnect ()
    {
        return pendingConnect != null;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected void canConnect ()
    {
        pendingConnect.run ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected boolean wantRead ()
    {
        return pendingRead != null;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected void canRead ()
    {
        pendingRead.run ();
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected boolean wantWrite ()
    {
        return pendingWrite != null;
    }

    /**
     * {@inheritDoc}
     */
    @Override // AbstractAsyncSelectableChannel
    protected void canWrite ()
    {
        pendingWrite.run ();
    }

    private class AsyncConnect implements Runnable, Handle
    {
        private final Callback <Void> callback;

        public AsyncConnect (Callback <Void> callback)
        {
            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.callback = callback;
        }

        @Override
        public void run ()
        {
            try
            {
                if (socketChannel.finishConnect ())
                {
                    pendingConnect = null;
                    callback.onSuccess (null);
                }
            }
            catch (IOException ex)
            {
                pendingConnect = null;
                callback.onError (ex);
            }
        }

        @Override
        public boolean cancel ()
        {
            return false; // It is always too late
        }
    }

    private class AsyncReadSingle implements Runnable, Handle
    {
        private final ByteBuffer dst;
        private final Callback <Integer> callback;

        public AsyncReadSingle (ByteBuffer dst, Callback <Integer> callback)
        {
            if (dst == null)
                throw new IllegalArgumentException ("Dst is null");

            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.dst = dst;
            this.callback = callback;
        }

        @Override
        public void run ()
        {
            try
            {
                int l = socketChannel.read (dst);

                if (l != 0 || dst.remaining () == 0)
                {
                    pendingRead = null;
                    callback.onSuccess (Integer.valueOf (l));
                }
            }
            catch (IOException ex)
            {
                pendingWrite = null;
                callback.onError (ex);
            }
        }

        @Override
        public boolean cancel ()
        {
            if (pendingRead == this)
            {
                pendingRead = null;
                updateInterestOps ();
                return true;
            }
            else return false;
        }
    }

    private class AsyncWriteSingle implements Runnable, Handle
    {
        private final ByteBuffer src;
        private final Callback <Integer> callback;

        public AsyncWriteSingle (ByteBuffer src, Callback <Integer> callback)
        {
            if (src == null)
                throw new IllegalArgumentException ("Src is null");

            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.src = src;
            this.callback = callback;
        }

        @Override
        public void run ()
        {
            try
            {
                int l = socketChannel.write (src);

                if (l != 0 || src.remaining () == 0)
                {
                    pendingWrite = null;
                    callback.onSuccess (Integer.valueOf (l));
                }
            }
            catch (IOException ex)
            {
                pendingWrite = null;
                callback.onError (ex);
            }
        }

        @Override
        public boolean cancel ()
        {
            if (pendingWrite == this)
            {
                pendingWrite = null;
                updateInterestOps ();
                return true;
            }
            else return false;
        }
    }

    private class AsyncReadMultiple implements Runnable, Handle
    {
        private final ByteBuffer [] dsts;
        private final int offset;
        private final int length;
        private final Callback <Long> callback;

        public AsyncReadMultiple (
            ByteBuffer [] dsts, int offset, int length,
            Callback <Long> callback)
        {
            if (dsts == null)
                throw new IllegalArgumentException ("Dsts is null");

            if (offset < 0)
                throw new IllegalArgumentException ("Offset (" + offset +
                    ") < 0");

            if (length < 0)
                throw new IllegalArgumentException ("Length (" + length +
                    ") < 0");

            int dstsLength = dsts.length;

            if (offset + length > dstsLength)
                throw new IllegalArgumentException ("Offset (" + offset +
                    ") + length (" + length + ") > dsts length (" + dstsLength +
                    ")");

            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.dsts = dsts.clone ();
            this.offset = offset;
            this.length = length;
            this.callback = callback;
        }

        @Override
        public void run ()
        {
            try
            {
                long l = socketChannel.read (dsts, offset, length);

                if (l != 0L || noRemaining ())
                {
                    pendingRead = null;
                    callback.onSuccess (Long.valueOf (l));
                }
            }
            catch (IOException ex)
            {
                pendingWrite = null;
                callback.onError (ex);
            }
        }

        @Override
        public boolean cancel ()
        {
            if (pendingRead == this)
            {
                pendingRead = null;
                updateInterestOps ();
                return true;
            }
            else return false;
        }

        private boolean noRemaining ()
        {
            for (ByteBuffer dst: dsts)
                if (dst.remaining () > 0) return false;

            return true;
        }
    }

    private class AsyncWriteMultiple implements Runnable, Handle
    {
        private final ByteBuffer [] srcs;
        private final int offset;
        private final int length;
        private final Callback <Long> callback;

        public AsyncWriteMultiple (
            ByteBuffer [] srcs, int offset, int length,
            Callback <Long> callback)
        {
            if (srcs == null)
                throw new IllegalArgumentException ("Srcs is null");

            if (offset < 0)
                throw new IllegalArgumentException ("Offset (" + offset +
                    ") < 0");

            if (length < 0)
                throw new IllegalArgumentException ("Length (" + length +
                    ") < 0");

            int srcsLength = srcs.length;

            if (offset + length > srcsLength)
                throw new IllegalArgumentException ("Offset (" + offset +
                    ") + length (" + length + ") > srcs length (" +
                    srcsLength + ")");

            if (callback == null)
                throw new IllegalArgumentException ("Callback is null");

            this.srcs = srcs.clone ();
            this.offset = offset;
            this.length = length;
            this.callback = callback;
        }

        @Override
        public void run ()
        {
            try
            {
                long l = socketChannel.write (srcs, offset, length);

                if (l != 0L || noRemaining ())
                {
                    pendingRead = null;
                    callback.onSuccess (Long.valueOf (l));
                }
            }
            catch (IOException ex)
            {
                pendingWrite = null;
                callback.onError (ex);
            }
        }

        @Override
        public boolean cancel ()
        {
            if (pendingRead == this)
            {
                pendingRead = null;
                updateInterestOps ();
                return true;
            }
            else return false;
        }

        private boolean noRemaining ()
        {
            for (ByteBuffer src: srcs)
                if (src.remaining () > 0) return false;

            return true;
        }
    }
}
